package com.atguigu.flink.chapter11;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.ZoneId;
import java.time.ZoneOffset;

public class Flink12_Time_Event_SQL {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
               env.setParallelism(1);

           StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
           // 设置本地时区
           tableEnv.getConfig().setLocalTimeZone(ZoneOffset.ofHours(0));

           //  bigint  -->  TIMESTAMP_LTZ(3)
           // 想要当作事件时间，还需要插入水印 添加事件时间
           // watermark for et as et - interval '3' second  et 作为水印时间， 减去 乱序程度 3秒
           tableEnv
                   .executeSql(" create table sensor(" +
                                   "  id string," +
                                   "  ts bigint," +
                                   "  vc int , " +
                                   "  et as to_timestamp_ltz(ts,3)," +
                                    "  watermark for et as et - interval '3' second" +
                                   ")with (" +
                                   "    'connector' = 'filesystem'," +
                                   "    'path'='input/sensor.txt'," +
                                   "    'format'='csv' " +
                                   ")");

           //    "  et2 as to_timestamp( from_unixtime(ts/1000))"  这个方法精度更低，不好用，就用上面那个
           tableEnv.from("sensor").printSchema();




          tableEnv.sqlQuery("select * from sensor").execute().print();








           }
}
